from absl import app, flags
import pandas as pd
import datetime

from coin.proto.coin_order_gateway_pb2 import OrderGatewayLog, OrderEvent
import coin.proto.coin_order_enums_pb2 as coin_order
from coin.proto.coin_strategy_pb2 import StrategyLog

from coin.base.param_util import split_string
from coin.base.timestamp import get_timestamp
from coin.tool.strat_monitor.util.monitor_util import get_topic_from_strat
from coin.util.queue.config import KafkaConfig
from coin.util.queue.tools.kafka_archive import run_from_kafka_archive
from coin.tool.strat_monitor.handlers.strategy_summary_calculator import (_get_accounting_currency, 
                                                                          get_fill_type_inferred)
from coin.support.pta.logic.pta_stats_calculator import calculate_turnover
from xunkemgmt_client.tool.slack_noti import send_to_slack
from coin.strategy.accounting.fee import fee_info, get_fee_rate
from coin.exchange.kr_rest.product.product_impl import generate_product_from_str2


FLAGS = flags.FLAGS

# binance fee discount rate in percentage
BINANCE_FEE_OFF = {
  'BNB_SPOT': 0.25,
  'BNB_FUTURES': 0.1,
}


class AccountSymbolChecker(object):
  def __init__(self):
    self._account_symbol = []

  def handle(self, partition, record):
    topic = record.topic
    pb = StrategyLog()
    pb.ParseFromString(record.value)
    if pb.type == StrategyLog.OG_LOG:
      self._handle_og_log(pb)

  def _handle_og_log(self, pb):
    assert pb.HasField('og_log'), ("log type %s has no og_log" % pb.type)
    og_log = pb.og_log
    if og_log.type == OrderGatewayLog.ORDER_EVENT:
      self._fetch_order_filled(og_log)

  def _from_native_currency(self, fee_currency):
    if fee_currency == 'BCH':
      fee_currency = 'BCHN'
    elif fee_currency == 'BSV':
      fee_currency = 'BCHSV'
    return fee_currency

  def _fetch_order_filled(self, og_log):
    if not og_log.HasField('account_request'):
      return
    order_event = og_log.event
    account_request = og_log.account_request
    if order_event.type == OrderEvent.ORDER_FILLED and order_event.fee and order_event.fee_currency:
      accounting_currency = _get_accounting_currency(account_request.market_type,
                                                     account_request.exchange,
                                                     order_event.symbol,
                                                     order_event.event_time)
      if order_event.fill_type == coin_order.UNKNOWN_FILL_TYPE: 
        order_event.fill_type = get_fill_type_inferred(order_event)
      if account_request.exchange == 'Binance':
        if (order_event.fill_type == coin_order.MAKER_FILL_TYPE and order_event.fee > 0) or \
          (order_event.fill_type == coin_order.TAKER_FILL_TYPE and order_event.fee < 0):
          order_event.fee = -order_event.fee
      fee_currency = self._from_native_currency(order_event.fee_currency)
      symbol_info = [account_request.market_type,
                     account_request.exchange,
                     account_request.owner,
                     order_event.symbol,
                     accounting_currency,
                     order_event.fill_type,
                     order_event.fill_qty,
                     order_event.fill_price,
                     order_event.fee,
                     fee_currency]
      self._account_symbol.append(symbol_info)

  def combine_account_symbol(self):
    df = pd.DataFrame(self._account_symbol, 
      columns=['market_type', 'exchange', 'owner', 'symbol', 'accounting_currency',
      'fill_type', 'fill_qty', 'fill_price', 'fee', 'fee_currency'])
    df.dropna(inplace=True)
    df.drop_duplicates(inplace=True)
    df.sort_values(df.columns.tolist(), inplace=True)
    df['maker_fee'], df['taker_fee'] = pd.Series(), pd.Series()
    df['maker_turnover_in_accy'], df['taker_turnover_in_accy'] = pd.Series(), pd.Series()
    for i, row in df.iterrows():
      if row['fee_currency'].upper() not in row['symbol'].upper():
        print(f'The fee rate of product {row["symbol"]} is unable to be calculated \
        by fee currency {row["fee_currency"]} for {row["market_type"]}-{row["exchange"]}')
        continue
      turnover_in_accy = calculate_turnover(market_type=row['market_type'],
                                            exchange=row['exchange'],
                                            symbol=row['symbol'],
                                            volume=row['fill_qty'],
                                            price=row['fill_price'])
      if row['fee_currency'] != row['accounting_currency'] or row['market_type'] == 'Options':
        fee = row['fee'] * row['fill_price']
      else:
        fee = row['fee']
      if row['fill_type'] == coin_order.MAKER_FILL_TYPE:
        df.at[i, 'maker_fee'] = fee
        df.at[i, 'maker_turnover_in_accy'] = turnover_in_accy
      elif row['fill_type'] == coin_order.TAKER_FILL_TYPE:
        df.at[i, 'taker_fee'] = fee
        df.at[i, 'taker_turnover_in_accy'] = turnover_in_accy
      df['maker_fee_rate'] = df['maker_fee'] / df['maker_turnover_in_accy']
      df['taker_fee_rate'] = df['taker_fee'] / df['taker_turnover_in_accy']
    df = df.drop(['accounting_currency', 'fill_type', 'fill_qty', 'fee', 'fee_currency'], axis=1)
    return df


def agg_cal_fee_rate(df):
  df_agg = df.groupby(
        ['market_type', 'exchange', 'owner', 'symbol'], as_index=False).sum(min_count=1)
  df_agg['maker_fee_rate'] = df_agg['maker_fee'] / df_agg['maker_turnover_in_accy']
  df_agg['taker_fee_rate'] = df_agg['taker_fee'] / df_agg['taker_turnover_in_accy']
  return df_agg


def filter_account_symbols(account_symbols, market_type=None, exchange=None, owner=None, symbol=None):
  if market_type:
    market_type_list = [elem.strip() for elem in market_type.split(',')]
    account_symbols = account_symbols[account_symbols['market_type'].isin(market_type_list)]
  if exchange:
    exchange_list = [elem.strip() for elem in exchange.split(',')]
    account_symbols = account_symbols[account_symbols['exchange'].isin(exchange_list)]
  if owner:
    owner_list = [elem.strip() for elem in owner.split(',')]
    account_symbols = account_symbols[account_symbols['owner'].isin(owner_list)]
  if symbol:
    symbol_list = [elem.strip() for elem in symbol.split(',')]
    account_symbols = account_symbols[account_symbols['symbol'].isin(symbol_list)]
  return account_symbols
  

def get_diff_order_registered_fee(df_order):
  output = []
  for _, row in df_order.iterrows():
    product = generate_product_from_str2(row['market_type'], row['exchange'], None, row['symbol'])
    maker_fee, taker_fee = get_fee_rate(row['market_type'], row['exchange'], product, row['owner'])
    if pd.notnull(row['maker_fee_rate']) and abs(row['maker_fee_rate'] - maker_fee) > 1e-6 or \
      pd.notnull(row['taker_fee_rate']) and abs(row['taker_fee_rate'] - taker_fee) > 1e-6:
      row = row.to_dict()
      row['registered_maker_fee_rate'] = maker_fee
      row['registered_taker_fee_rate'] = taker_fee
      output.append(row)
  df_output = pd.DataFrame(output)
  return df_output


def apply_binance_discount(df_diff):
  fee_rate_diff_list = []
  for _, fee_rate in df_diff.iterrows():
    verified = False
    if fee_rate['exchange'] == 'Binance':
      fee_ratio = round(fee_rate['registered_taker_fee_in_bp'] /
                        fee_rate['taker_fee_in_bp'], 2)
      if fee_rate['market_type'] == 'Spot':
        verified = (fee_ratio == 1 - BINANCE_FEE_OFF['BNB_SPOT'])
      elif fee_rate['market_type'] == 'Futures':
        verified = (fee_ratio == 1 - BINANCE_FEE_OFF['BNB_FUTURES'])
    if not verified:
      fee_rate_diff_list.append(fee_rate)
  df_output = pd.DataFrame(fee_rate_diff_list, columns=df_diff.columns)
  return df_output


def main(_):
  kafka_config = None
  if FLAGS.kafka_config_filename is not None:
    kafka_config = KafkaConfig.from_cmd_config(FLAGS.kafka_config_filename)
  topics = get_topic_from_strat(strat_list=None, kafka_config=kafka_config, active=True)
  end_ts = get_timestamp()
  offset_nanosec = int(datetime.timedelta(hours=FLAGS.offset).total_seconds() * 10**9)
  start_ts = end_ts - offset_nanosec
  handler = AccountSymbolChecker()
  run_from_kafka_archive(topics=topics,
                         kafka_config=kafka_config,
                         timestamp_from=start_ts,
                         timestamp_to=end_ts,
                         callback=handler.handle)
  account_symbols = handler.combine_account_symbol()
  df_agg = agg_cal_fee_rate(account_symbols)
  df = df_agg[['market_type', 'exchange', 'owner', 'symbol', 'maker_fee_rate', 'taker_fee_rate']]
  filtered_df = filter_account_symbols(df,
                  market_type=FLAGS.market_type,
                  exchange=FLAGS.exchange,
                  owner=FLAGS.owner,
                  symbol=FLAGS.symbol)
  df_output = get_diff_order_registered_fee(filtered_df)
  if len(df_output) > 0:
    for col in ('maker_fee_rate', 'taker_fee_rate', 'registered_maker_fee_rate', 'registered_taker_fee_rate'):
      df_output[col] = df_output[col].apply(lambda x: x * 1e4)
    col_names = [col_name.replace("fee_rate", "fee_in_bp") for col_name in df_output.columns]
    df_output.columns = col_names
  if FLAGS.binance_fee_off:
    df_output = apply_binance_discount(df_output)
  print(df_output.to_string())
  if FLAGS.slack_receiver is not None:
    send_to_slack("```" +  "fetch time range: " + str(start_ts) + " - " + str(end_ts) + 
      "\nfee rate diff between order event and registered:\n" + 
      df_output.to_string(na_rep='', float_format='%.2f', index=False) + "```", FLAGS.slack_receiver, 'msg')


if __name__ == '__main__':
  flags.DEFINE_string('kafka_config_filename',
                      '../../coin_deploy/support_monitor/config/kafka_config.json',
                      'kafka config')
  flags.DEFINE_string('market_type', None, 'market_type.')
  flags.DEFINE_string('exchange', None, 'exchange.')
  flags.DEFINE_string('owner', None, 'owner.')
  flags.DEFINE_string('symbol', None, 'symbol.')
  flags.DEFINE_float('offset', 1.0, 'offset.')
  flags.DEFINE_string('slack_receiver', None, 'slack_receiver')
  flags.DEFINE_boolean('binance_fee_off', False, 'apply binance fee discounts plan')
  app.run(main)
